package kafkaex

import (
	"errors"
	"fmt"

	"github.com/IBM/sarama"
	"github.com/google/uuid"
)

// NewKafkaManager new a kafka mq manager
func NewKafkaManager(brokers ...string) *KafkaManager {
	m := &KafkaManager{
		brokers: brokers,
	}
	return m
}

// KafkaManager kafka mq manager
type KafkaManager struct {
	brokers   []string
	config    *sarama.Config
	client    sarama.Client        // sarama kafka sdk client
	consumers map[string]IConsumer // consumer group map
}

// init init
func (m *KafkaManager) Init() error {
	config := sarama.NewConfig()
	// 同一个消费组中消费者订阅不同的topic,那么需要Sticky策略，该策略会使得分配更加均匀
	config.Consumer.Group.Rebalance.Strategy = sarama.NewBalanceStrategySticky()
	// TODO: CONFIGS
	m.config = config
	client, err := sarama.NewClient(m.brokers, m.config)
	if err != nil {
		return err
	}
	m.client = client
	return nil
}

func (m *KafkaManager) NewConsumer(groupID string, topics ...string) error {
	// 根据client创建consumerGroup
	group, err := sarama.NewConsumerGroupFromClient(groupID, m.client)
	if err != nil {
		return err
	}
	id := uuid.NewString()
	if m.consumers == nil {
		m.consumers = map[string]IConsumer{}
	}
	c := &Consumer{
		ready: make(chan bool),
	}
	c.id = id
	c.group = group
	c.topics = topics
	m.consumers[c.id] = c
	return nil
}

func (m *KafkaManager) ConsumersGo() {
	for _, c := range m.consumers {
		_, err := c.Go()
		if err != nil {
			println(err.Error())
		}
	}
}

func (m *KafkaManager) ConsumerGo(id string) error {
	if m.consumers == nil {
		return errors.New("no found consumers")
	}
	consumer, ok := m.consumers[id]
	if ok {
		return fmt.Errorf("no found consumer %v", id)
	}
	_, err := consumer.Go()
	return err
}
